package com.dy.dws.DocRegisterWin;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.dy.realtime.common.base.BaseApp;
import com.dy.realtime.common.bean.DwsDocRegisterBean;
import com.dy.realtime.common.bean.DwsTradeHospPatGenAGPresBean;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.DateFormatUtil;
import com.dy.realtime.common.util.DorisMapFunction;
import com.dy.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.dws.DocRegisterWin
 * @className: DwsDocRegisterWin
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/16 19:04
 * @version: 1.0
 */

public class DwsDocRegisterWin extends BaseApp {
    public static void main(String[] args) {
        new DwsDocRegisterWin().start(Constant.TOPIC_DWD_doctor_register, Constant.TOPIC_DWS_TRADE_REFUND_DOC_REGISTER_WIN, 4, 10027);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
//         dataStreamSource.print();
        SingleOutputStreamOperator<DwsDocRegisterBean> etlStream = getEtlStream(dataStreamSource);
//        etlStream.print();
        SingleOutputStreamOperator<DwsDocRegisterBean> result = getReduce(etlStream);
//        result.print();
        result.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_doc_register_win"));
    }

    private static SingleOutputStreamOperator<DwsDocRegisterBean> getReduce(SingleOutputStreamOperator<DwsDocRegisterBean> etlStream) {
        return etlStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsDocRegisterBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<DwsDocRegisterBean>() {
                    @Override
                    public long extractTimestamp(DwsDocRegisterBean dwsDocRegisterBean, long l) {
                        return dwsDocRegisterBean.getTs();
                    }
                })).windowAll(TumblingEventTimeWindows.of(Time.seconds(2)))
                .process(new ProcessAllWindowFunction<DwsDocRegisterBean, DwsDocRegisterBean, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<DwsDocRegisterBean, DwsDocRegisterBean, TimeWindow>.Context context, Iterable<DwsDocRegisterBean> iterable, Collector<DwsDocRegisterBean> collector) throws Exception {
                        ArrayList<DwsDocRegisterBean> dwsDocRegisterBeans = ListUtil.toList(iterable);
                        long sum = dwsDocRegisterBeans.stream().mapToLong(e -> e.getRegisterCount()).sum();
                        String start = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String end = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                        DwsDocRegisterBean next = dwsDocRegisterBeans.iterator().next();
                        collector.collect(DwsDocRegisterBean.builder()
                                        .id(next.getId())
                                        .birthday(next.getBirthday())
                                        .consultationFee(next.getConsultationFee())
                                        .curDate(curDate)
                                        .edt(end)
                                        .stt(start)
                                        .genderCode(next.getGenderCode())
                                        .registerTime(next.getRegisterTime())
                                        .registerCount(sum)
                                        .hospitalId(next.getHospitalId())
                                        .specialtyCode(next.getSpecialtyCode())
                                        .titleCode(next.getTitleCode())
                                .build());
                    }
                });
    }

    private static SingleOutputStreamOperator<DwsDocRegisterBean> getEtlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, DwsDocRegisterBean>() {
            @Override
            public void flatMap(String s, Collector<DwsDocRegisterBean> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String register_time = jsonObject.getString("register_time");
                if (register_time != null) {
                    DwsDocRegisterBean dwsDocRegisterBean = JSONObject.parseObject(jsonObject.toString(), DwsDocRegisterBean.class);
                    dwsDocRegisterBean.setRegisterCount(1L);
                    collector.collect(dwsDocRegisterBean);
                }
            }
        });
    }
}
